/*
 *
 *  * | Licensed 未经许可不能去掉「OPENIITA」相关版权
 *  * +----------------------------------------------------------------------
 *  * | Author: xw2sy@163.com
 *  * +----------------------------------------------------------------------
 *
 *  Copyright [2024] [OPENIITA]
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 * /
 */

package cc.iotkit.plugins.wzr.service;

import cc.iotkit.common.utils.StringUtils;
import cc.iotkit.common.utils.ThreadUtil;
import cc.iotkit.common.utils.UniqueIdUtil;
import cc.iotkit.plugin.core.IPlugin;
import cc.iotkit.plugin.core.IPluginConfig;
import cc.iotkit.plugin.core.thing.IThingService;
import cc.iotkit.plugin.core.thing.actions.DeviceState;
import cc.iotkit.plugin.core.thing.actions.IDeviceAction;
import cc.iotkit.plugin.core.thing.actions.down.ServiceInvoke;
import cc.iotkit.plugin.core.thing.actions.up.*;
import cc.iotkit.plugin.core.thing.model.ThingDevice;
import cc.iotkit.plugins.wzr.conf.MqttConfig;
import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.bean.copier.CopyOptions;
import cn.hutool.core.util.IdUtil;
import com.gitee.starblues.bootstrap.annotation.AutowiredType;
import com.gitee.starblues.bootstrap.realize.PluginCloseListener;
import com.gitee.starblues.core.PluginCloseType;
import com.gitee.starblues.core.PluginInfo;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.vertx.core.Future;
import io.vertx.core.Vertx;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import io.vertx.mqtt.MqttClient;
import io.vertx.mqtt.MqttClientOptions;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.*;

/**
 * @author sjg
 */
@Slf4j
@Service
public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable {

    @Autowired
    private PluginInfo pluginInfo;
    @Autowired
    private MqttConfig mqttConfig;

    @Autowired
    @AutowiredType(AutowiredType.Type.MAIN_PLUGIN)
    private IPluginConfig pluginConfig;

    @Autowired
    @AutowiredType(AutowiredType.Type.MAIN_PLUGIN)
    private IThingService thingService;

    @Autowired
    private AuthVerticle authVerticle;

    @Autowired
    private MqttDevice mqttDevice;

    private final ScheduledThreadPoolExecutor emqxConnectTask = ThreadUtil.newScheduled(1, "emqx_connect");

    private Vertx vertx;
    private String deployedId;

    private MqttClient client;

    private boolean mqttConnected = false;

    private boolean authServerStarted = false;


    public static final Map<String, Set<String>> CLIENT_DEVICE_MAP = new HashMap<>();
    public static final Map<String, TopoNet> GATEWAYMAC_PVID_MAP = new ConcurrentHashMap<>();

    @PostConstruct
    public void init() {
        vertx = Vertx.vertx();
        try {
            //获取插件最新配置替换当前配置
            Map<String, Object> config = pluginConfig.getConfig(pluginInfo.getPluginId());
            BeanUtil.copyProperties(config, mqttConfig, CopyOptions.create().ignoreNullValue());
            String serverPassword = IdUtil.fastSimpleUUID();
            MqttClientOptions options = new MqttClientOptions()
                    .setClientId("server")
                    .setUsername("server")
                    .setPassword("123456")
                    .setCleanSession(true)
                    .setMaxInflightQueue(100)
                    .setKeepAliveInterval(60);
            if (mqttConfig.isSsl()) {
                options.setSsl(true)
                        .setTrustAll(true);
            }
            client = MqttClient.create(vertx, options);
            mqttDevice.setClient(client);

            authVerticle.setPort(mqttConfig.getAuthPort());
            authVerticle.setServerPassword("123456");

            emqxConnectTask.scheduleWithFixedDelay(this, 3, 3, TimeUnit.SECONDS);
        } catch (Throwable e) {
            log.error("mqtt plugin startup error", e);
        }
    }
    @Override
    public void run() {
        if (!authServerStarted) {
            try {
                CountDownLatch countDownLatch = new CountDownLatch(1);
                Future<String> future = vertx.deployVerticle(authVerticle);
                future.onSuccess((s -> {
                    deployedId = s;
                    countDownLatch.countDown();
                    authServerStarted = true;
                    log.info("start emqx auth plugin success");
                }));
                future.onFailure(e -> {
                    countDownLatch.countDown();
                    authServerStarted = false;
                    log.error("start emqx auth plugin failed", e);
                });
                countDownLatch.await();
            } catch (Exception e) {
                authServerStarted = false;
                log.error("start emqx auth server failed", e);
            }
        }

        if (mqttConnected) {
            return;
        }

        try {
            String[] topics = mqttConfig.getTopics().split(",");
            Map<String, Integer> subscribes = new HashMap<>(topics.length);
            for (String topic : topics) {
                subscribes.put(topic, 1);
            }
            client.connect(mqttConfig.getPort(), mqttConfig.getHost(), s -> {
                if (s.succeeded()) {
                    log.info("client connect success.");
                    mqttConnected = true;
                    client.subscribe(subscribes, e -> {
                        if (e.succeeded()) {
                            log.info("===>subscribe success: {}", e.result());
                        } else {
                            log.error("===>subscribe fail: ", e.cause());
                        }
                    });

                } else {
                    mqttConnected = false;
                    log.error("client connect fail: ", s.cause());
                }
            }).publishHandler(msg -> {
                String topic = msg.topicName();

                if (topic.contains("/c/")) {
                    return;
                }
                JsonObject payload = msg.payload().toJsonObject();
                log.info("Client received message on [{}] payload [{}] with QoS [{}]", topic, payload, msg.qosLevel());
                /**
                 * 此处不做过多业务处理，只做数据获取上报和平台通知，具体业务由AV去处理
                 *  网关上线离线心跳
                 *  组网三步通知
                 *  设备增加删除
                 *  设备上线，下线，心跳
                 *  群组增加，删除，修改
                 *  定时任务增加，删除，修改
                 */
                try {
                    IDeviceAction action = null;
                    int firstSlashIndex = topic.indexOf('/');
                    int secondSlashIndex = topic.indexOf('/', firstSlashIndex + 1);
                    String wzrgwdv = topic.substring(firstSlashIndex + 1, secondSlashIndex);
                    ThingDevice wzrgw = thingService.getDevice(wzrgwdv);
                    if (wzrgw == null) {
                        return;
                    }
                    //上报网关在线状态,网关有数据就要报在线
                    online(wzrgw.getProductKey(),wzrgw.getDeviceName());
                    if (topic.contains("/devices/me/attributes/request/1")) {
                        //查询场景
                        if (payload.getString("_version-scene")!=null && payload.getString("_version-scene").equals("_version-scene")) {
                            //回复，不过网关好像没走这里来过
                        }
                        //处理网关心跳
                       if (payload.getString("sharedKeys").equals("_version-pvid,_version-pgmac,_version-scene,_version-linkage,_version-calendar,_version-mvid,_version-mgmac")) {
                           online(wzrgw.getProductKey(),wzrgw.getDeviceName());
                           return;
                       }
                    }
                    //处理网关业务
                    if (topic.contains("/devices/me/telemetry")) {
                        JsonObject tool = payload.getJsonObject("tool");
                       if (tool != null) {
                           String fcn = tool.getString("fcn");
                           if (StringUtils.isBlank(fcn)) {
                               return;
                           }

                            //网关开始搜索设备
//                           if (fcn.equalsIgnoreCase("netSearch")){
//                               return;
//                           }
                           //读取已经发现的设备信息

                           if (fcn.equalsIgnoreCase("netNum")) {
                               JsonObject output = tool.getJsonObject("output");
                               Integer toponum = output.getInteger("toponum");
                               action = PropertyReport.builder()
                                       .params(Map.of(
                                               "toponum", toponum
                                       ))
                                       .build();
                               action.setId(UniqueIdUtil.newRequestId());
                               action.setProductKey(wzrgw.getProductKey());
                               action.setDeviceName(wzrgw.getDeviceName());
                               action.setTime(System.currentTimeMillis());
                               thingService.post(pluginInfo.getPluginId(), action);
                               // 返回节点数量时, 重置拓扑网络信息
                               TopoNet topoNet = new TopoNet(toponum, new CopyOnWriteArrayList<>());
                               GATEWAYMAC_PVID_MAP.put(wzrgw.getDeviceName(), topoNet);
                               // 根据节点数量, 循环下发获取节点信息, 每次下发获取十个
                               /*for (int i = 1; i <= toponum; i += 10) {
                                   int start = i;
                                   int cnt = 10;
                                   // 调度器异步延迟执行, 每个延迟loop秒执行, 实际效果每个下发间隔2秒
                                   int loop = ((i - 1) / 10 + 1) * 2;
                                   emqxConnectTask.schedule(() -> {
                                       mqttDevice.serviceInvoke(ServiceInvoke.builder()
                                               .id(UniqueIdUtil.newRequestId())
                                               .productKey(wzrgw.getProductKey())
                                               .deviceName(wzrgw.getDeviceName())
                                               .name("selectNet")
                                               .params(Map.of(
                                                       "start", start,
                                                       "cnt", cnt
                                               ))
                                               .build());
                                       log.info("下发获取节点信息 -> gw:{}, start:{}, cnt:{}", wzrgw.getDeviceName(), start, cnt);
                                   }, loop, TimeUnit.SECONDS);
                               }*/
                               return;
                           }
                           if (fcn.equalsIgnoreCase("netInfo")){
                               JsonObject output = tool.getJsonObject("output");
                               JsonArray macInfoArray = output.getJsonArray("macInfo");
                               for (int i = 0; i < macInfoArray.size(); i++) {
                                   log.error("注册子设备"+macInfoArray.getJsonObject(i).getString("mac"));
                                   //注册设备，但是未入网，MT-L20是微自然光感器的产商proKey，MN-T20S是微自然视觉传感器
                                   //子设备注册, 当前微自然单灯固定产品key=hATm2Tswh3TDik8W, 微自然光感器key=wzrguangganqi, 微自然视觉传感器key=wzraishijue
                                   String platformProKey;
                                   String platformModel;
                                   if ("MT-L20".equalsIgnoreCase(macInfoArray.getJsonObject(i).getString("proKey"))) {
                                       platformProKey="wzrguangganqi";
                                       platformModel="wzr_ggq";
                                   } else if ("MN-T20S".equalsIgnoreCase(macInfoArray.getJsonObject(i).getString("proKey"))) {
                                       platformProKey="wzraishijue";
                                       platformModel="wzr_aisj";
                                   } else {
                                       platformProKey="hATm2Tswh3TDik8W";
                                       platformModel="wzr_ddc";
                                   }
                                   String mac = macInfoArray.getJsonObject(i).getString("mac");
                                   thingService.post(
                                           pluginInfo.getPluginId(),
                                           fillAction(
                                                   SubDeviceRegister.builder()
                                                           .productKey(wzrgw.getProductKey())
                                                           .deviceName(wzrgw.getDeviceName())
                                                           .version("1.0")
                                                           .subs(List.of(
                                                                   DeviceRegister.builder()
                                                                           .productKey(platformProKey)
                                                                           .deviceName(mac)
                                                                           .model(platformModel)
                                                                           .version("1.0")
                                                                           .build()
                                                           ))
                                                           .build()
                                           )
                                   );

                                   //PLC 设备添加 添加设备 pvid
                                   TopoNet topoNet = GATEWAYMAC_PVID_MAP.get(wzrgw.getDeviceName());
                                   if (topoNet == null || topoNet.getTopoInfoList().contains(mac)) {
                                       // 重复上报节点mac时，则不处理
                                       return;
                                   }

                                   topoNet.getTopoInfoList().add(mac);
                                   if (topoNet.getTopoNum() == topoNet.getTopoInfoList().size()) {
                                       // 节点数量与集合中存储的mac数量一致，则下发网关设置pvid
                                       for (int idx = 0; idx < topoNet.getTopoInfoList().size(); idx++) {
                                           int pvid = idx + 1;
                                           String devmac = topoNet.getTopoInfoList().get(idx);
                                           // 调度器异步延迟执行, 每个延迟idx秒执行, 实际效果每个下发间隔1秒
                                           emqxConnectTask.schedule(() -> {
                                               mqttDevice.serviceInvoke(ServiceInvoke.builder()
                                                       .id(UniqueIdUtil.newRequestId())
                                                       .productKey(wzrgw.getProductKey())
                                                       .deviceName(wzrgw.getDeviceName())
                                                       .name("pvid")
                                                       .params(Map.of(
                                                               "gwmac", pvid,
                                                               "devmac", devmac
                                                       ))
                                                       .build());
                                               log.info("下发 pvid 成功 -> gw:{}, dev:{}, pvid:{}", wzrgw.getDeviceName(), devmac, pvid);
                                           }, idx, TimeUnit.SECONDS);
                                       }
                                   }
                               }
                               return;
                           }
                           if (fcn.equalsIgnoreCase("topoInfo")) {
                               JsonObject output = tool.getJsonObject("output");
                               JsonArray macInfoArray = output.getJsonArray("topoInfo");
                               for (int i = 0; i < macInfoArray.size(); i++) {
                                   String macInfo = macInfoArray.getString(i);
                                   // macInfo为"TOPOINFO:C20241202434,2,1,1,1,31,36,1"需截取成C20241202434, 即第一个:和第一个,之间的内容
                                   String mac = macInfo.substring(macInfo.indexOf(":") + 1, macInfo.indexOf(","));
                                   if (wzrgw.getDeviceName().equalsIgnoreCase(mac)) {
                                       // 节点为网关设备时, 不作处理
                                       return;
                                   }
                                   log.info("注册微自然子设备: {}", mac);
                                   //注册设备，但是未入网
                                   //子设备注册
                                   thingService.post(
                                           pluginInfo.getPluginId(),
                                           fillAction(
                                                   SubDeviceRegister.builder()
                                                           .productKey(wzrgw.getProductKey())
                                                           .deviceName(wzrgw.getDeviceName())
                                                           .version("1.0")
                                                           .subs(List.of(
                                                                   DeviceRegister.builder()
                                                                           //当前微自然单灯固定产品key= hATm2Tswh3TDik8W
                                                                           .productKey("hATm2Tswh3TDik8W")
                                                                           .deviceName(mac)
                                                                           .model("wzr_ddc")
                                                                           .version("1.0")
                                                                           .build()
                                                           ))
                                                           .build()
                                           )
                                   );
                                   /*thingService.post(
                                       pluginInfo.getPluginId(),
                                       fillAction(
                                               DeviceRegister.builder()
                                               //当前的固定单灯key= hATm2Tswh3TDik8W
                                               .productKey("hATm2Tswh3TDik8W")
                                               .deviceName(macInfoArray.getJsonObject(i).getString("mac"))
                                               .model("wzr_ddc")
                                               .version("1.0")
                                               .build()
                                       )
                                   );*/
                                   //PLC 设备添加 添加设备 pvid
                                   TopoNet topoNet = GATEWAYMAC_PVID_MAP.get(wzrgw.getDeviceName());
                                   if (topoNet == null || topoNet.getTopoInfoList().contains(mac)) {
                                       // 重复上报节点mac时，则不处理
                                       return;
                                   }

                                   topoNet.getTopoInfoList().add(mac);
                                   if (topoNet.getTopoNum() == topoNet.getTopoInfoList().size()) {
                                       // 节点数量与集合中存储的mac数量一致，则下发网关设置pvid
                                       for (int idx = 0; idx < topoNet.getTopoInfoList().size(); idx++) {
                                           int pvid = idx + 1;
                                           String devmac = topoNet.getTopoInfoList().get(idx);
                                           // 调度器异步延迟执行, 每个延迟idx秒执行, 实际效果每个下发间隔1秒
                                           emqxConnectTask.schedule(() -> {
                                               mqttDevice.serviceInvoke(ServiceInvoke.builder()
                                                       .id(UniqueIdUtil.newRequestId())
                                                       .productKey(wzrgw.getProductKey())
                                                       .deviceName(wzrgw.getDeviceName())
                                                       .name("pvid")
                                                       .params(Map.of(
                                                               "gwmac", pvid,
                                                               "devmac", devmac
                                                       ))
                                                       .build());
                                               log.info("下发 pvid 成功 -> gw:{}, dev:{}, pvid:{}", wzrgw.getDeviceName(), devmac, pvid);
                                           }, idx, TimeUnit.SECONDS);
                                       }
                                   }


                               }
                           }
                       }
                    }
                    //增加设备，修改，删除
                    if (topic.contains("/devices/me/attributes")) {
                        //由AV下发成功为准
                    }
                    //子设备上线
                    if (topic.contains("/gateway/connect")) {
                        String device = payload.getString("device", "");
                        if (StringUtils.isBlank(device)) {
                            return;
                        }
                        online(thingService.getDevice(device).getProductKey(),device);
                    }
                    //子设备下线
                    if (topic.contains("/gateway/disconnect")) {
                        String device = payload.getString("device", "");
                        if (StringUtils.isBlank(device)) {
                            return;
                        }
                        offline(device);
                    }
                    //子设备心跳保活和属性上报，默认五分钟
                    if (topic.contains("/gateway/telemetry")) {
                        if (payload.getString("mac") != null && payload.getString("mac").length() > 0) {
                            String mac = payload.getString("mac");

                            JsonArray deinfolist = payload.getJsonArray(payload.getString("mac"));
                            JsonObject device = deinfolist.getJsonObject(0);
                            //如果key没有对应的设备，直接return
                            if (thingService.getDevice(mac) == null) {
                                return;
                            }
                            //心跳保活(hb心跳保活, 说明书只有5分钟, 但实际网关硬件有出现过超过30分钟的给设备心跳保活, 目前改为属性上报保活)
                            /*if (device.containsKey("hb")) {
                                ThingDevice thingDevice = thingService.getDevice(mac);
                                online(thingDevice.getProductKey(), mac);
                            }*/
                            /*if (device.containsKey("cfg")) {
                                online("hATm2Tswh3TDik8W",mac);
                            }*/
                            //属性上报
                            if (device.containsKey("dp")) {
                                ThingDevice thingDevice = thingService.getDevice(mac);
                                online(thingDevice.getProductKey(), mac);
                                JsonObject defParams = JsonObject.mapFrom(new HashMap<>(0));
                                JsonObject params = device.getJsonObject("dp", defParams);
                                action = PropertyReport.builder()
                                        .params(params.getMap())
                                        .build();
                                action.setId(UniqueIdUtil.newRequestId());
                                action.setProductKey(thingDevice.getProductKey());
                                action.setDeviceName(thingDevice.getDeviceName());
                                action.setTime(System.currentTimeMillis());
                                thingService.post(pluginInfo.getPluginId(), action);
                                return;
                            }
                        }
                    }
                    //群组，定时业务监听下发成功且收到即可，业务后续在再处理

                } catch (Exception e) {
                    log.error("message is illegal.", e);
                }
            }).closeHandler(e -> {
                mqttConnected = false;
                log.info("client closed");
            }).exceptionHandler(event -> log.error("client fail", event));
        } catch (Exception e) {
            log.error("start emqx client failed", e);
        }

    }


    public ThingDevice getDevice(String topic) {
        String[] topicParts = topic.split("/");
        if (topicParts.length < 3) {
            return null;
        }
        return ThingDevice.builder()
                .productKey(topicParts[2])
                .deviceName(topicParts[3])
                .build();
    }

    public void online(String pk, String dn) {
        //上线
        thingService.post(
                pluginInfo.getPluginId(),
                fillAction(DeviceStateChange.builder()
                        .productKey(pk)
                        .deviceName(dn)
                        .state(DeviceState.ONLINE)
                        .build()
                )
        );
    }

    public void offline(String devicename) {
        thingService.post(
                pluginInfo.getPluginId(),
                fillAction(DeviceStateChange.builder()
                        .productKey(thingService.getDevice(devicename).getProductKey())
                        .deviceName(devicename)
                        .state(DeviceState.OFFLINE)
                        .build()
                )
        );
    }


    private IDeviceAction fillAction(IDeviceAction action) {
        action.setId(UniqueIdUtil.newRequestId());
        action.setTime(System.currentTimeMillis());
        return action;
    }

    /**
     * 回复设备
     */
    private void reply(String topic, JsonObject payload, int code) {
        Map<String, Object> payloadReply = new HashMap<>();
        payloadReply.put("id", payload.getString("id"));
        payloadReply.put("method", payload.getString("method") + "_reply");
        payloadReply.put("code", code);
        payloadReply.put("data", payload.getJsonObject("params"));
        topic = topic.replace("/s/", "/c/") + "_reply";

        String finalTopic = topic;
        client.publish(topic, JsonObject.mapFrom(payloadReply).toBuffer(), MqttQoS.AT_LEAST_ONCE, false, false)
                .onSuccess(h -> {
                    log.info("publish {} success", finalTopic);
                });
    }

    public void replyT(String topic, JsonObject payload, int code) {
        client.publish(topic, JsonObject.mapFrom(payload).toBuffer(), MqttQoS.AT_LEAST_ONCE, false, false)
                .onSuccess(h -> {
                    log.info("publish {} success", topic);
                });
    }

    @Override
    public void close(GenericApplicationContext applicationContext, PluginInfo pluginInfo, PluginCloseType closeType) {
        try {
            log.info("plugin close,type:{},pluginId:{}", closeType, pluginInfo.getPluginId());
            if (deployedId != null) {
                CountDownLatch wait = new CountDownLatch(1);
                Future<Void> future = vertx.undeploy(deployedId);
                future.onSuccess(unused -> {
                    log.info("emqx plugin stopped success");
                    wait.countDown();
                });
                future.onFailure(h -> {
                    log.error("emqx plugin stopped failed", h);
                    wait.countDown();
                });
                wait.await(5, TimeUnit.SECONDS);
            }

            client.disconnect()
                    .onSuccess(unused -> {
                        mqttConnected = false;
                        log.info("stop emqx connect success");
                    })
                    .onFailure(unused -> log.error("stop emqx connect failure"));

            emqxConnectTask.shutdown();

        } catch (Throwable e) {
            log.error("emqx plugin stop error", e);
        }
    }

    @Override
    public Map<String, Object> getLinkInfo(String pk, String dn) {
        return null;
    }
}
